/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.catalina.tribes.transport.bio;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Arrays;

import org.apache.catalina.tribes.RemoteProcessException;
import org.apache.catalina.tribes.io.XByteBuffer;
import org.apache.catalina.tribes.transport.AbstractSender;
import org.apache.catalina.tribes.transport.Constants;
import org.apache.catalina.tribes.transport.DataSender;
import org.apache.catalina.tribes.transport.SenderState;
import org.apache.catalina.tribes.util.StringManager;

/**
 * Send cluster messages with only one socket. Ack and keep Alive Handling is
 * supported
 * 
 * @author Peter Rossbach
 * @author Filip Hanik
 * @version $Id: BioSender.java 939515 2010-04-29 23:59:49Z kkolinko $
 * @since 5.5.16
 */
public class BioSender extends AbstractSender implements DataSender {

	private static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory
			.getLog(BioSender.class);

	/**
	 * The string manager for this package.
	 */
	protected static StringManager sm = StringManager
			.getManager(Constants.Package);

	// ----------------------------------------------------- Instance Variables

	/**
	 * The descriptive information about this implementation.
	 */
	private static final String info = "DataSender/3.0";

	/**
	 * current sender socket
	 */
	private Socket socket = null;
	private OutputStream soOut = null;
	private InputStream soIn = null;

	protected XByteBuffer ackbuf = new XByteBuffer(
			Constants.ACK_COMMAND.length, true);

	// ------------------------------------------------------------- Constructor

	public BioSender() {
	}

	// ------------------------------------------------------------- Properties

	/**
	 * Return descriptive information about this implementation and the
	 * corresponding version number, in the format
	 * <code>&lt;description&gt;/&lt;version&gt;</code>.
	 */
	public String getInfo() {
		return (info);
	}

	// --------------------------------------------------------- Public Methods

	/**
	 * Connect other cluster member receiver
	 * 
	 * @see org.apache.catalina.tribes.transport.IDataSender#connect()
	 */
	public void connect() throws IOException {
		openSocket();
	}

	/**
	 * disconnect and close socket
	 * 
	 * @see IDataSender#disconnect()
	 */
	public void disconnect() {
		boolean connect = isConnected();
		closeSocket();
		if (connect) {
			if (log.isDebugEnabled())
				log
						.debug(sm.getString("IDataSender.disconnect",
								getAddress().getHostAddress(), new Integer(
										getPort()), new Long(0)));
		}

	}

	/**
	 * Send message
	 * 
	 * @see org.apache.catalina.tribes.transport.IDataSender#sendMessage(,
	 *      ChannelMessage)
	 */
	public void sendMessage(byte[] data, boolean waitForAck) throws IOException {
		IOException exception = null;
		setAttempt(0);
		try {
			// first try with existing connection
			pushMessage(data, false, waitForAck);
		} catch (IOException x) {
			SenderState.getSenderState(getDestination()).setSuspect();
			exception = x;
			if (log.isTraceEnabled())
				log.trace(sm.getString("IDataSender.send.again", getAddress()
						.getHostAddress(), new Integer(getPort())), x);
			while (getAttempt() < getMaxRetryAttempts()) {
				try {
					setAttempt(getAttempt() + 1);
					// second try with fresh connection
					pushMessage(data, true, waitForAck);
					exception = null;
				} catch (IOException xx) {
					exception = xx;
					closeSocket();
				}
			}
		} finally {
			setRequestCount(getRequestCount() + 1);
			keepalive();
			if (exception != null)
				throw exception;
		}
	}

	/**
	 * Name of this SockerSender
	 */
	public String toString() {
		StringBuffer buf = new StringBuffer("DataSender[(");
		buf.append(super.toString()).append(")");
		buf.append(getAddress()).append(":").append(getPort()).append("]");
		return buf.toString();
	}

	// --------------------------------------------------------- Protected
	// Methods

	/**
	 * open real socket and set time out when waitForAck is enabled is socket
	 * open return directly
	 */
	protected void openSocket() throws IOException {
		if (isConnected())
			return;
		try {
			socket = new Socket();
			InetSocketAddress sockaddr = new InetSocketAddress(getAddress(),
					getPort());
			socket.connect(sockaddr, (int) getTimeout());
			socket.setSendBufferSize(getTxBufSize());
			socket.setReceiveBufferSize(getRxBufSize());
			socket.setSoTimeout((int) getTimeout());
			socket.setTcpNoDelay(getTcpNoDelay());
			socket.setKeepAlive(getSoKeepAlive());
			socket.setReuseAddress(getSoReuseAddress());
			socket.setOOBInline(getOoBInline());
			socket.setSoLinger(getSoLingerOn(), getSoLingerTime());
			socket.setTrafficClass(getSoTrafficClass());
			setConnected(true);
			soOut = socket.getOutputStream();
			soIn = socket.getInputStream();
			setRequestCount(0);
			setConnectTime(System.currentTimeMillis());
			if (log.isDebugEnabled())
				log
						.debug(sm.getString("IDataSender.openSocket",
								getAddress().getHostAddress(), new Integer(
										getPort()), new Long(0)));
		} catch (IOException ex1) {
			SenderState.getSenderState(getDestination()).setSuspect();
			if (log.isDebugEnabled())
				log.debug(sm.getString("IDataSender.openSocket.failure",
						getAddress().getHostAddress(), new Integer(getPort()),
						new Long(0)), ex1);
			throw (ex1);
		}

	}

	/**
	 * close socket
	 * 
	 * @see DataSender#disconnect()
	 * @see DataSender#closeSocket()
	 */
	protected void closeSocket() {
		if (isConnected()) {
			if (socket != null) {
				try {
					socket.close();
				} catch (IOException x) {
				} finally {
					socket = null;
					soOut = null;
					soIn = null;
				}
			}
			setRequestCount(0);
			setConnected(false);
			if (log.isDebugEnabled())
				log
						.debug(sm.getString("IDataSender.closeSocket",
								getAddress().getHostAddress(), new Integer(
										getPort()), new Long(0)));
		}
	}

	/**
	 * Push messages with only one socket at a time Wait for ack is needed and
	 * make auto retry when write message is failed. After sending error close
	 * and reopen socket again.
	 * 
	 * After successfull sending update stats
	 * 
	 * WARNING: Subclasses must be very carefull that only one thread call this
	 * pushMessage at once!!!
	 * 
	 * @see #closeSocket()
	 * @see #openSocket()
	 * @see #writeData(ChannelMessage)
	 * 
	 * @param data
	 *            data to send
	 * @since 5.5.10
	 */

	protected void pushMessage(byte[] data, boolean reconnect,
			boolean waitForAck) throws IOException {
		keepalive();
		if (reconnect)
			closeSocket();
		if (!isConnected())
			openSocket();
		soOut.write(data);
		soOut.flush();
		if (waitForAck)
			waitForAck();
		SenderState.getSenderState(getDestination()).setReady();

	}

	/**
	 * Wait for Acknowledgement from other server FIXME Please, not wait only
	 * for three charcters, better control that the wait ack message is correct.
	 * 
	 * @param timeout
	 * @throws java.io.IOException
	 * @throws java.net.SocketTimeoutException
	 */
	protected void waitForAck() throws java.io.IOException {
		try {
			boolean ackReceived = false;
			boolean failAckReceived = false;
			ackbuf.clear();
			int bytesRead = 0;
			int i = soIn.read();
			while ((i != -1) && (bytesRead < Constants.ACK_COMMAND.length)) {
				bytesRead++;
				byte d = (byte) i;
				ackbuf.append(d);
				if (ackbuf.doesPackageExist()) {
					byte[] ackcmd = ackbuf.extractDataPackage(true).getBytes();
					ackReceived = Arrays
							.equals(
									ackcmd,
									org.apache.catalina.tribes.transport.Constants.ACK_DATA);
					failAckReceived = Arrays
							.equals(
									ackcmd,
									org.apache.catalina.tribes.transport.Constants.FAIL_ACK_DATA);
					ackReceived = ackReceived || failAckReceived;
					break;
				}
				i = soIn.read();
			}
			if (!ackReceived) {
				if (i == -1)
					throw new IOException(sm.getString("IDataSender.ack.eof",
							getAddress(), new Integer(socket.getLocalPort())));
				else
					throw new IOException(sm.getString("IDataSender.ack.wrong",
							getAddress(), new Integer(socket.getLocalPort())));
			} else if (failAckReceived && getThrowOnFailedAck()) {
				throw new RemoteProcessException(
						"Received a failed ack:org.apache.catalina.tribes.transport.Constants.FAIL_ACK_DATA");
			}
		} catch (IOException x) {
			String errmsg = sm.getString("IDataSender.ack.missing",
					getAddress(), new Integer(socket.getLocalPort()), new Long(
							getTimeout()));
			if (SenderState.getSenderState(getDestination()).isReady()) {
				SenderState.getSenderState(getDestination()).setSuspect();
				if (log.isWarnEnabled())
					log.warn(errmsg, x);
			} else {
				if (log.isDebugEnabled())
					log.debug(errmsg, x);
			}
			throw x;
		} finally {
			ackbuf.clear();
		}
	}
}
